package org.example;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.time.DayOfWeek;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;

public class Task2Driver1 {
    public static class Task2Mapper1 extends Mapper<Object,Text,Text,Text>{
        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] parts=value.toString().split("\t");
            if(parts.length>=2){
                String dateString = parts[0];
                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd"); // 定义日期格式
                // 将字符串转换为LocalDate对象
                LocalDate date = LocalDate.parse(dateString, formatter);
                // 获取星期几
                DayOfWeek dayOfWeek = date.getDayOfWeek();
                Text weekday=new Text(dayOfWeek.toString());
                Text result=new Text(parts[1]);
                context.write(weekday, result);
            }
        }
    }

    public static class Task2Reducer1 extends Reducer<Text,Text,Text,Text>{
        @Override
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            long sumPurchase = 0;
            long sumRedeem = 0;
            long count=0;
            for (Text val : values) {
                String[] tmpval=val.toString().split(",");
                long purchase = Integer.parseInt(tmpval[0]);
                long redeem = Integer.parseInt(tmpval[1]);
                sumPurchase += purchase;
                sumRedeem += redeem;
                count++;
            }
            double average1= (double) sumPurchase /count;
            double average2= (double) sumRedeem /count;
            long intAverage1=Math.round(average1);
            long intAverage2=Math.round(average2);
            String result=String.valueOf(intAverage1)+","+String.valueOf(intAverage2);
            Text textResult=new Text(result);
            context.write(key, textResult);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "task2 amounts/weekday");

        job.setJarByClass(Task2Driver1.class);
        job.setMapperClass(Task2Driver1.Task2Mapper1.class);
        job.setReducerClass(Task2Driver1.Task2Reducer1.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
